<?php

namespace App\Http\Controller\Test;

use Swoft\Http\Server\Annotation\Mapping\Controller;
use Swoft\Http\Server\Annotation\Mapping\RequestMapping;
use Swoft\Http\Server\Annotation\Mapping\RequestMethod;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Swoft\Log\Helper\CLog;
use Swoft\Log\Helper\Log;

/**
 * Class MqConsumerController
 * @Controller()
 */
class MqConsumerController extends BaseController {

    /**
     * 消费端 消费端需要保持运行状态实现方式使用supervisor进行监控
     * 以daemon守护进程开启的supervisor无法起到守护作用
     * @param $channel 信道
     * @param $connection  连接
     **/
    function shutdown($channel, $connection)
    {
        //关闭信道
        $channel->close();
        //关闭连接
        $connection->close();
        //写入日志
//        Log::info('closed %s log', 'rabbitmq-consumer');
    }

    //消息处理
    function process_message($message)
    {
        //休眠一秒
        sleep(1);
        //打印消息
        vdump($message->body."n");
        //自定义错误为rabbitmq
//        Log::info($message->body.' %s log', 'rabbitmq-consumer');

        //手动发送ack：需【$channel->basic_consume第四个参数为false】
//        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
//        // Send a message with the string "quit" to cancel the consumer.
//        if ($message->body === 'quit') {
//            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
//        }
    }

    /**
     * 启动商品消费者
     * @RequestMapping("/consumer/goods",method={RequestMethod::GET})
     * @throws Throwable
     */
    public function goods()
    {
        $param = config('AMQP');
        $amqpDetail = config('goods_queue');
        $connection = new AMQPStreamConnection(
            $param['host'],
            $param['port'],
            $param['login'],
            $param['password'],
            $param['vhost']
        );
        //连接信道
        $channel = $connection->channel();
        //流量控制
        $channel->basic_qos(0, 1, false);
        //创建交换机(Exchange)
        $channel->exchange_declare($amqpDetail['exchange_name'], $amqpDetail['exchange_type'], false, true, false);
        //创建队列
        $channel->queue_declare($amqpDetail['queue_name'], false, true, false, false);
        //绑定队列和交换机
        $channel->queue_bind($amqpDetail['queue_name'], $amqpDetail['exchange_name']);
        //消费消息
        $channel->basic_consume($amqpDetail['queue_name'], '', false, true, false, false, array($this, 'process_message'));
        //退出，执行shutdown来关闭通道与连接
        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        // 阻塞队列监听事件
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }

    /**
     * 启动订单消费者
     * @RequestMapping("/consumer/order",method={RequestMethod::GET})
     * @throws Throwable
     */
    public function order()
    {
        $param = config('AMQP');
        $amqpDetail = config('order_queue');
        $connection = new AMQPStreamConnection(
            $param['host'],
            $param['port'],
            $param['login'],
            $param['password'],
            $param['vhost']
        );
        //连接信道
        $channel = $connection->channel();
        //流量控制
        $channel->basic_qos(0, 1, false);
        //创建交换机(Exchange)
        $channel->exchange_declare($amqpDetail['exchange_name'], $amqpDetail['exchange_type'], false, true, false);
        //创建队列
        $channel->queue_declare($amqpDetail['queue_name'], false, true, false, false);
        //绑定队列和交换机【可foreach循环 base.php -> topic_queue -> route_key】
        $channel->queue_bind($amqpDetail['queue_name'], $amqpDetail['exchange_name'],$amqpDetail['route_key']);
        //消费消息
        $channel->basic_consume($amqpDetail['queue_name'], '', false, true, false, false, array($this, 'process_message'));
        //退出，执行shutdown来关闭通道与连接
        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        // 阻塞队列监听事件
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }


}